package com.company.transmqtt.service;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.Arrays;
import java.util.List;

@Service
public class MqttForwardService {

    private static final Logger log = LogManager.getLogger(MqttForwardService.class);
    @Value("${mqtt.source.topics}")
    private String sourceTopics;

    @Value("${mqtt.target.topic.prefix.src}")
    private String targetTopicSrc;
    @Value("${mqtt.target.topic.prefix.dst}")
    private String targetTopicDst;

    @Autowired
    @Qualifier("sourceMqttClientFactory")
    private MqttPahoClientFactory sourceClientFactory;

    @Autowired
    @Qualifier("targetMqttClientFactory")
    private MqttPahoClientFactory targetClientFactory;

    @Autowired
    @Qualifier("sourceClientId")
    private String sourceClientId;

    @Autowired
    @Qualifier("targetClientId")
    private String targetClientId;

    private MqttPahoMessageHandler targetMessageHandler;
    private MessageHandler sourceMessageHandler;

    @PostConstruct
    public void init() {
        // 初始化目标MQTT消息处理器（用于发送消息到目标broker）
        targetMessageHandler = new MqttPahoMessageHandler(targetClientId, targetClientFactory);
        targetMessageHandler.setAsync(true);
        targetMessageHandler.setDefaultTopic("default/topic");
        targetMessageHandler.afterPropertiesSet();

        // 初始化源MQTT消息处理器（用于接收消息）
        sourceMessageHandler = new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                try {
                    String topic = (String) message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC);
                    Object payloadObj = message.getPayload();
                    byte[] payload;
                    
                    log.info("Received MQTT message on topic: {}", topic);
                    
                    if (payloadObj instanceof byte[]) {
                        payload = (byte[]) payloadObj;
                    } else if (payloadObj instanceof String) {
                        payload = ((String) payloadObj).getBytes();
                    } else {
                        log.error("Unsupported payload type: {}", payloadObj.getClass().getName());
                        return;
                    }
                    
                    Integer qos = (Integer) message.getHeaders().get(MqttHeaders.QOS);
                    Boolean retained = (Boolean) message.getHeaders().get(MqttHeaders.RETAINED);
                    
                    if (topic != null) {
                        log.info("Payload length: {}, QoS: {}, Retained: {}", payload.length, qos, retained);
                        forwardMessage(topic, payload, qos != null ? qos : 1, retained != null ? retained : false);
                    } else {
                        log.error("Invalid message: topic");
                    }
                } catch (Exception e) {
                    log.error("Error processing MQTT message: {}", e.getMessage());
                }
            }
        };

        log.info("MQTT Forward Service initialized");
        log.info("Listening to topics: {}", sourceTopics);
        log.info("Target topic prefix: {} => {}", targetTopicSrc, targetTopicDst);
    }

    private void forwardMessage(String sourceTopic, byte[] payload, int qos, boolean retained) {
        try {
            // 构建目标topic（可选添加前缀）
            String targetTopic = sourceTopic.replace(targetTopicSrc, targetTopicDst);
            
            // 创建转发消息
            Message<byte[]> message = MessageBuilder
                    .withPayload(payload)
                    .setHeader(MqttHeaders.TOPIC, targetTopic)
                    .setHeader(MqttHeaders.QOS, qos)
                    .setHeader(MqttHeaders.RETAINED, retained)
                    .build();
            
            // 发送到目标broker
            targetMessageHandler.handleMessage(message);
            
            log.info("Forwarded message from:{} to: {}, payload length: {} bytes", sourceTopic, targetTopic,  + payload.length);
            
        } catch (Exception e) {
            log.error("Error forwarding message: {}", e.getMessage());
        }
    }

    public MessageHandler getSourceMessageHandler() {
        return sourceMessageHandler;
    }

    public List<String> getSourceTopics() {
        return Arrays.asList(sourceTopics.split(","));
    }

    @PreDestroy
    public void destroy() {
        if (targetMessageHandler != null) {
            targetMessageHandler.destroy();
        }
        log.info("MQTT Forward Service destroyed");
    }
}